package com.alarm.ecp.utils;


import com.alarm.ecp.pojo.DeviceData;
import com.alarm.ecp.pojo.EventPO;
import com.alibaba.fastjson2.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.Objects;
import java.util.Properties;

public class KafkaUtil {
    private static final Logger logger = LoggerFactory.getLogger(KafkaUtil.class);
    public static DataStream<EventPO> read(String[] args, StreamExecutionEnvironment env) throws IOException {
        // 从命令行参数获取配置文件路径
        ParameterTool parameterTool = ParameterTool.fromArgs(args);

        String configFilePath = parameterTool.get("configPath");
        if (configFilePath == null || configFilePath.isEmpty()) {
            throw new IllegalArgumentException("Missing configPath parameter.");
        }

        // 从指定路径加载 properties 文件
        ParameterTool config = ParameterTool.fromPropertiesFile(new File(configFilePath));

        // 配置 Kafka 消费者属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "polygon-fence-group");

        // 创建 Kafka 消费者
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<>("zhongji-qianyi-real-time-fence", new SimpleStringSchema(), properties);
        kafkaConsumer.setStartFromLatest();

        DataStream<String> inputKafkaStream = env.addSource(kafkaConsumer);

        // 解析 JSON 数据流
        DataStream<EventPO> parsedDataStream = inputKafkaStream.map((MapFunction<String, EventPO>) value -> {
            try {
                return JSONObject.parseObject(value, EventPO.class);
            } catch (Exception e) {
                logger.error(e.getMessage());
            }
            return null;
        }).filter(Objects::nonNull);

        return parsedDataStream;
    }
}
